/*
    SUSAN® - Sole of Unix Save ANything

      Copyright (C) 2011-2016 Skyatlas Co. LTD

   

   
*/
/*
 * Originally written by Kern Sibbald for inclusion in apcupsd,
 * but heavily modified for SUSAN
 */
/*
 * @file
 * tcp server code
 */

#include "include/susan.h"
#include "lib/alist.h"
#include "lib/berrno.h"
#include "lib/bnet_server_tcp.h"
#include "lib/bsock_tcp.h"
#include "lib/bsys.h"
#include "lib/thread_list.h"
#include "lib/watchdog.h"

#include <netinet/in.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <arpa/inet.h>
#include <netdb.h>
#ifdef HAVE_ARPA_NAMESER_H
#include <arpa/nameser.h>
#endif
#ifdef HAVE_RESOLV_H
//#include <resolv.h>
#endif

#ifdef HAVE_POLL_H
#include <poll.h>
#elif HAVE_SYS_POLL_H
#include <sys/poll.h>
#endif

#include <algorithm>
#include <atomic>
#include <array>
#include <vector>

static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static std::atomic<bool> quit{false};

struct s_sockfd {
  int fd;
  int port;
};

/**
 * Stop the Threaded Network Server if its realy running in a separate thread.
 * e.g. set the quit flag and wait for the other thread to exit cleanly.
 */
void BnetStopAndWaitForThreadServerTcp(pthread_t tid)
{
  Dmsg0(100, "BnetThreadServer: Request Stop\n");
  quit = true;
  if (!pthread_equal(tid, pthread_self())) {
    Dmsg0(100, "BnetThreadServer: Wait until finished\n");
    pthread_join(tid, nullptr);
    Dmsg0(100, "BnetThreadServer: finished\n");
  }
}

/**
 * Perform a cleanup for the Threaded Network Server check if there is still
 * something to do or that the cleanup already took place.
 */
static void CleanupBnetThreadServerTcp(alist* sockfds, ThreadList& thread_list)
{
  Dmsg0(100, "CleanupBnetThreadServerTcp: start\n");

  if (sockfds && !sockfds->empty()) {
    s_sockfd* fd_ptr = (s_sockfd*)sockfds->first();
    while (fd_ptr) {
      close(fd_ptr->fd);
      fd_ptr = (s_sockfd*)sockfds->next();
    }
    sockfds->destroy();
  }

  if (!thread_list.ShutdownAndWaitForThreadsToFinish()) {
    Emsg1(M_ERROR, 0, _("Could not destroy thread list.\n"));
  }
  Dmsg0(100, "CleanupBnetThreadServerTcp: finish\n");
}

class BNetThreadServerCleanupObject {
 public:
  BNetThreadServerCleanupObject(alist* sockfds, ThreadList& thread_list)
      : sockfds_(sockfds), thread_list_(thread_list)
  {
  }

  ~BNetThreadServerCleanupObject()
  {
    CleanupBnetThreadServerTcp(sockfds_, thread_list_);
  }

 private:
  alist* sockfds_;
  ThreadList& thread_list_;
};

static void RemoveDuplicateAddresses(dlist* addr_list)
{
  IPADDR* ipaddr = nullptr;
  IPADDR* next = nullptr;
  IPADDR* to_free = nullptr;

  for (ipaddr = (IPADDR*)addr_list->first(); ipaddr;
       ipaddr = (IPADDR*)addr_list->next(ipaddr)) {
    next = (IPADDR*)addr_list->next(ipaddr);
    while (next) {
      /*
       * See if the addresses match.
       */
      if (ipaddr->GetSockaddrLen() == next->GetSockaddrLen() &&
          memcmp(ipaddr->get_sockaddr(), next->get_sockaddr(),
                 ipaddr->GetSockaddrLen()) == 0) {
        to_free = next;
        next = (IPADDR*)addr_list->next(next);
        addr_list->remove(to_free);
        delete to_free;
      } else {
        next = (IPADDR*)addr_list->next(next);
      }
    }
  }
}

static void LogAllAddresses(dlist* addr_list)
{
  std::vector<char> buf(256 * addr_list->size());

  Dmsg1(100, "Addresses %s\n",
        BuildAddressesString(addr_list, buf.data(), buf.size()));
}

static int OpenSocketAndBind(IPADDR* ipaddr,
                             dlist* addr_list,
                             uint16_t port_number)
{
  int fd = -1;
  int tries = 0;

  do {
    ++tries;
    if ((fd = socket(ipaddr->GetFamily(), SOCK_STREAM, 0)) < 0) {
      Bmicrosleep(10, 0);
    }
  } while (fd < 0 && tries < 6);

  if (fd < 0) {
    BErrNo be;
    std::array<char, 256> buf1;
    std::vector<char> buf2(256 * addr_list->size());

    Emsg3(M_ABORT, 0,
          _("Cannot open stream socket. ERR=%s. Current %s All %s\n"),
          be.bstrerror(), ipaddr->build_address_str(buf1.data(), buf1.size()),
          BuildAddressesString(addr_list, buf2.data(), buf2.size()));

    return -1;
  }

  int reuseaddress = 1;
  if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (sockopt_val_t)&reuseaddress,
                 sizeof(reuseaddress)) < 0) {
    BErrNo be;
    Emsg1(M_WARNING, 0, _("Cannot set SO_REUSEADDR on socket: %s\n"),
          be.bstrerror());
    return -2;
  }

  tries = 0;

  do {
    ++tries;
    if (bind(fd, ipaddr->get_sockaddr(), ipaddr->GetSockaddrLen()) < 0) {
      BErrNo be;
      Emsg2(M_WARNING, 0, _("Cannot bind port %d: ERR=%s: Retrying ...\n"),
            ntohs(port_number), be.bstrerror());
      Bmicrosleep(5, 0);
    } else {
      // success
      return fd;
    }
  } while (tries < 3);

  return -3;
}

/**
 * Become Threaded Network Server
 *
 * This function is able to handle multiple server ips in
 * ipv4 and ipv6 style. The Addresse are give in a comma
 * separated string in bind_addr
 *
 * At the moment it is impossible to bind to different ports.
 */
void BnetThreadServerTcp(
    dlist* addr_list,
    int max_clients,
    alist* sockfds,
    ThreadList& thread_list,
    std::function<void*(ConfigurationParser* config, void* bsock)>
        HandleConnectionRequest,
    ConfigurationParser* config,
    std::atomic<BnetServerState>* const server_state,
    std::function<void*(void* bsock)> UserAgentShutdownCallback,
    std::function<void()> CustomCallback)
{
  BNetThreadServerCleanupObject cleanup_object(sockfds, thread_list);

  quit = false;  // allow other threads to set this true during initialization
  if (server_state) { server_state->store(BnetServerState::kStarting); }

  RemoveDuplicateAddresses(addr_list);
  LogAllAddresses(addr_list);

#ifdef HAVE_POLL
  nfds_t number_of_filedescriptors = 0;
#endif

  IPADDR* ipaddr = nullptr;

  foreach_dlist (ipaddr, addr_list) {
    s_sockfd* fd_ptr = (s_sockfd*)alloca(sizeof(s_sockfd));

    fd_ptr->port = ipaddr->GetPortNetOrder();
    fd_ptr->fd = OpenSocketAndBind(ipaddr, addr_list, fd_ptr->port);

    if (fd_ptr->fd < 0) {
      BErrNo be;
      Emsg2(M_ERROR, 0, _("Cannot bind port %d: ERR=%s.\n"),
            ntohs(fd_ptr->port), be.bstrerror());
      if (server_state) { server_state->store(BnetServerState::kError); }
      return;
    }

    listen(fd_ptr->fd, 50); /* tell system we are ready */
    sockfds->append(fd_ptr);

#ifdef HAVE_POLL
    number_of_filedescriptors++;
#endif
  }

  thread_list.Init(max_clients, HandleConnectionRequest,
                   UserAgentShutdownCallback);

#ifdef HAVE_POLL
  struct pollfd* pfds =
      (struct pollfd*)alloca(sizeof(struct pollfd) * number_of_filedescriptors);
  memset(pfds, 0, sizeof(struct pollfd) * number_of_filedescriptors);

  int events = POLLIN;
#if defined(POLLRDNORM)
  events |= POLLRDNORM;
#endif
#if defined(POLLRDBAND)
  events |= POLLRDBAND;
#endif
#if defined(POLLPRI)
  events |= POLLPRI;
#endif

  s_sockfd* fd_ptr = nullptr;
  int i = 0;

  foreach_alist (fd_ptr, sockfds) {
    pfds[i].fd = fd_ptr->fd;
    pfds[i].events = events;
    i++;
  }
#endif

  if (server_state) { server_state->store(BnetServerState::kStarted); }

  while (!quit) {
    if (CustomCallback) { CustomCallback(); }
#ifndef HAVE_POLL
    int maxfd = 0;
    fd_set sockset;
    FD_ZERO(&sockset);

    s_sockfd* fd_ptr = nullptr;
    foreach_alist (fd_ptr, sockfds) {
      FD_SET((unsigned)fd_ptr->fd, &sockset);
      maxfd = std::max(fd_ptr->fd, maxfd);
    }

    struct timeval timeout {
      .tv_sec = 1, .tv_usec = 0
    };

    errno = 0;
    int status = select(maxfd + 1, &sockset, NULL, NULL, &timeout);

    if (status == 0) {
      continue;  // timeout: check if thread should quit
    } else if (status < 0) {
      BErrNo be; /* capture errno */
      if (errno == EINTR) { continue; }
      if (server_state) { server_state->store(BnetServerState::kError); }
      Emsg1(M_FATAL, 0, _("Error in select: %s\n"), be.bstrerror());
      break;
    }

    foreach_alist (fd_ptr, sockfds) {
      if (FD_ISSET(fd_ptr->fd, &sockset)) {
#else
    static constexpr int timeout_ms{1000};

    errno = 0;
    int status = poll(pfds, number_of_filedescriptors, timeout_ms);

    if (status == 0) {
      continue;  // timeout: check if thread should quit
    } else if (status < 0) {
      BErrNo be; /* capture errno */
      if (errno == EINTR) { continue; }
      Emsg1(M_FATAL, 0, _("Error in poll: %s\n"), be.bstrerror());

      break;
    }

    int cnt = 0;
    foreach_alist (fd_ptr, sockfds) {
      if (pfds[cnt++].revents & events) {
#endif

        int newsockfd = -1;
        socklen_t clilen;
        struct sockaddr_storage cli_addr; /* client's address */

        do {
          clilen = sizeof(cli_addr);
          newsockfd = accept(fd_ptr->fd, reinterpret_cast<sockaddr*>(&cli_addr),
                             &clilen);
        } while (newsockfd < 0 && errno == EINTR);
        if (newsockfd < 0) { continue; }

        int keepalive = 1;
        if (setsockopt(newsockfd, SOL_SOCKET, SO_KEEPALIVE,
                       (sockopt_val_t)&keepalive, sizeof(keepalive)) < 0) {
          BErrNo be;
          Emsg1(M_WARNING, 0, _("Cannot set SO_KEEPALIVE on socket: %s\n"),
                be.bstrerror());
        }

        /*
         * See who client is. i.e. who connected to us.
         */
        char buf[128];

        P(mutex);
        SockaddrToAscii(reinterpret_cast<sockaddr*>(&cli_addr), buf,
                        sizeof(buf));
        V(mutex);

        SusanSocket* bs;
        bs = new SusanSocketTCP;

        bs->fd_ = newsockfd;
        bs->SetWho(strdup("client"));
        bs->SetHost(strdup(buf));
        bs->SetPort(ntohs(fd_ptr->port));
        memset(&bs->peer_addr, 0, sizeof(bs->peer_addr));
        memcpy(&bs->client_addr, &cli_addr, sizeof(bs->client_addr));

        if (!thread_list.CreateAndAddNewThread(config, bs)) {
          Jmsg1(NULL, M_ABORT, 0, _("Could not add thread to list.\n"));
        }
      }
    }
  }
  if (server_state) { server_state->store(BnetServerState::kEnded); }
}
